package es

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/elastic/go-elasticsearch/v7"
	"github.com/elastic/go-elasticsearch/v7/esapi"
	"reflect"
	"strings"
	"sync"
)

var (
	once sync.Once
	es   *Es
)

type Es struct {
	client     *elasticsearch.Client  //es链接实例
	indexName  string                 //索引名称
	query      map[string]interface{} //查询构造器
	limit      *int                   //每页查询的数量
	offet      *int                   //查询偏移
	order      []string               //排序方式
	field      []string               //查询的字段
	lightField []string               //高亮字段
}

func New() *Es {
	once.Do(func() {
		cfg := elasticsearch.Config{
			Addresses: []string{
				"http://192.168.77.129:9200",
			},
		}
		client, err := elasticsearch.NewClient(cfg)
		if err != nil {
			fmt.Println(err)
		}
		res, err := client.Info()
		if err != nil {
			fmt.Println(err.Error())
		}
		if res.IsError() {
			fmt.Println(err.Error())
		}
		limit := 20
		es = &Es{
			client: client,
			limit:  &limit,
		}
	})
	return es
}

// 设置索引
func (p *Es) Index(indexName string) *Es {
	p.indexName = indexName
	return p
}

// 设置查询条件
func (p *Es) Query(query map[string]interface{}) *Es {
	p.query = query
	return p
}

// 设置查询每页条数
func (p *Es) Limit(limit int) *Es {
	p.limit = &limit
	return p
}

// 设置分页
func (p *Es) Page(page int) *Es {
	offet := (page - 1) * (*p.limit)
	p.offet = &offet
	return p
}

// 设置排序 如 age:asc
func (p *Es) Order(order ...string) *Es {
	p.order = order
	return p
}

// 设置查询返回的字段
func (p *Es) Field(field string) *Es {
	p.field = strings.Split(field, ",")
	return p
}

// 关键词高亮
func (p *Es) HighLight(field ...string) *Es {
	p.lightField = field
	return p
}

// 创建索引
func (p *Es) CreateIndex() error {
	req := esapi.IndicesCreateRequest{
		Index: p.indexName,
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return err
	}
	defer res.Body.Close()

	if res.IsError() {
		return nil
	}
	return nil
}

// 删除索引
func (p *Es) DeleteIndex() error {
	req := esapi.IndicesDeleteRequest{
		Index: []string{p.indexName},
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return err
	}
	defer res.Body.Close()

	if res.IsError() {
		return errors.New(res.String())
	}
	return nil
}

// 刷新索引
func (p *Es) RefreshIndex() error {
	req := esapi.IndicesRefreshRequest{
		Index: []string{p.indexName},
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return err
	}
	defer res.Body.Close()

	if res.IsError() {
		return errors.New(res.String())
	}
	return nil
}

// 创建字段映射
func (p *Es) CreateMapping(mapping map[string]interface{}) error {
	data := make(map[string]interface{})
	data["properties"] = mapping
	docJSON, err := json.Marshal(data)
	if err != nil {
		return err
	}
	req := esapi.IndicesPutMappingRequest{
		Index: []string{p.indexName},
		Body:  bytes.NewBuffer(docJSON),
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return err
	}
	defer res.Body.Close()

	if res.IsError() {
		return errors.New(res.String())
	}
	return nil
}

// 查看字段映射
func (p *Es) GetMapping() (interface{}, error) {
	req := esapi.IndicesGetMappingRequest{
		Index: []string{p.indexName},
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return "", err
	}
	defer res.Body.Close()

	if res.IsError() {
		return "", errors.New(res.String())
	}
	var mappingResponse map[string]interface{}
	err = json.NewDecoder(res.Body).Decode(&mappingResponse)
	if err != nil {
		return "", err
	}
	properties := mappingResponse[p.indexName].(map[string]interface{})["mappings"].(map[string]interface{})["properties"]
	return properties, nil
}

// 添加数据
func (p *Es) Insert(doc interface{}) error {
	body, _ := json.Marshal(doc)
	req := esapi.IndexRequest{
		Index: p.indexName,
		Body:  bytes.NewBuffer(body),
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return err
	}
	defer res.Body.Close()
	if res.IsError() {
		return errors.New(res.String())
	}
	return nil
}

// 批量新增数据
func (p *Es) InsertALL(documents interface{}) error {
	value := reflect.ValueOf(documents)
	if value.Kind() != reflect.Slice {
		return errors.New("documents must be a slice")
	}
	var body strings.Builder
	for i := 0; i < value.Len(); i++ {
		doc := value.Index(i).Interface()
		indexAction := map[string]interface{}{
			"index": map[string]interface{}{
				"_index": p.indexName,
			},
		}
		content, _ := json.Marshal(indexAction)
		docBytes, err := json.Marshal(doc)
		if err != nil {
			return err
		}
		body.WriteString(fmt.Sprintf("%s\n", string(content)))
		body.WriteString(fmt.Sprintf("%s\n", docBytes))
	}
	req := esapi.BulkRequest{
		Index:   p.indexName,
		Body:    strings.NewReader(body.String()),
		Refresh: "true",
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return err
	}
	defer res.Body.Close()
	if res.IsError() {
		return errors.New(res.String())
	}
	return nil
}

// 根据docid修改数据
func (p *Es) Update(docId string, data interface{}) error {
	body := make(map[string]interface{})
	body["doc"] = data
	content, _ := json.Marshal(body)
	req := esapi.UpdateRequest{
		Index:      p.indexName,
		DocumentID: docId,
		Body:       bytes.NewBuffer(content),
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return err
	}
	defer res.Body.Close()
	if res.IsError() {
		return errors.New(res.String())
	}
	return nil
}

// 根据条件修改数据 updateBuildString 构建参数 ctx._source.field = params.value
func (p *Es) UpdateByQuery(updateBuildString string, data interface{}) error {
	body := map[string]interface{}{
		"script": map[string]interface{}{
			"source": updateBuildString,
			"lang":   "painless",
			"params": data,
		},
	}
	for k, v := range p.query {
		body[k] = v
	}
	docJSON, err := json.Marshal(body)
	if err != nil {
		return err
	}
	req := esapi.UpdateByQueryRequest{
		Index: []string{p.indexName},
		Body:  bytes.NewBuffer(docJSON),
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return err
	}
	defer res.Body.Close()
	if res.IsError() {
		return errors.New(res.String())
	}
	return nil
}

// 搜索文档
func (p *Es) Find() (map[string]interface{}, error) {
	var queryBytes []byte
	if len(p.query) > 0 {
		if len(p.lightField) > 0 {
			lightFields := make(map[string]interface{})
			for _, v := range p.lightField {
				lightFields[v] = map[string]interface{}{}
			}
			p.query["highlight"] = map[string]interface{}{
				"fields": lightFields,
			}
		}
		queryBytes, _ = json.Marshal(p.query)
	}
	req := esapi.SearchRequest{
		Index:  []string{p.indexName},
		Body:   bytes.NewBuffer(queryBytes),
		Size:   p.limit,
		From:   p.offet,
		Sort:   p.order,
		Source: p.field,
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()
	if res.IsError() {
		return nil, errors.New(res.String())
	}
	var result map[string]interface{}
	if err1 := json.NewDecoder(res.Body).Decode(&result); err1 != nil {
		return nil, err1
	}
	return result["hits"].(map[string]interface{}), nil
}

// 根据docId查询一条文档
func (p *Es) FindOne(docId string) (map[string]interface{}, error) {
	req := esapi.GetRequest{
		Index:      p.indexName,
		DocumentID: docId,
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()
	if res.IsError() {
		return nil, errors.New(res.String())
	}
	var result map[string]interface{}
	if err1 := json.NewDecoder(res.Body).Decode(&result); err1 != nil {
		return nil, err1
	}
	return result, nil
}

// 根据docId删除一条文档
func (p *Es) DeleteOne(docId string) error {
	req := esapi.DeleteRequest{
		Index:      p.indexName,
		DocumentID: docId,
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return err
	}
	defer res.Body.Close()
	if res.IsError() {
		return errors.New(res.String())
	}
	return nil
}

// 根据条件删除文档
func (p *Es) Delete() error {
	queryBytes, _ := json.Marshal(p.query)
	refresh := true
	req := esapi.DeleteByQueryRequest{
		Index:   []string{p.indexName},
		Body:    bytes.NewBuffer(queryBytes),
		Refresh: &refresh,
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return err
	}
	defer res.Body.Close()
	if res.IsError() {
		return errors.New(res.String())
	}
	return nil
}

// 查询文档总数
func (p *Es) Count() (int, error) {
	var queryBytes []byte
	if len(p.query) > 0 {
		queryBytes, _ = json.Marshal(p.query)
	}
	req := esapi.CountRequest{
		Index: []string{p.indexName},
		Body:  bytes.NewBuffer(queryBytes),
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return 0, err
	}
	defer res.Body.Close()
	if res.IsError() {
		return 0, errors.New(res.String())
	}
	var result map[string]interface{}
	if err1 := json.NewDecoder(res.Body).Decode(&result); err1 != nil {
		return 0, err1
	}
	return result["count"].(int), nil
}

// 求和
func (p *Es) Sum(field string) (float64, error) {
	aggregationQuery := map[string]interface{}{
		"size": 0, // 不返回文档结果，只返回聚合结果
		"aggs": map[string]interface{}{
			"sum_field": map[string]interface{}{
				"sum": map[string]interface{}{
					"field": field,
				},
			},
		},
	}
	for k, v := range p.query {
		aggregationQuery[k] = v
	}

	queryBytes, _ := json.Marshal(aggregationQuery)
	req := esapi.SearchRequest{
		Index: []string{p.indexName},
		Body:  bytes.NewBuffer(queryBytes),
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return 0, err
	}
	defer res.Body.Close()
	if res.IsError() {
		return 0, errors.New(res.String())
	}
	var result map[string]interface{}
	if err1 := json.NewDecoder(res.Body).Decode(&result); err1 != nil {
		return 0, err1
	}
	sumValue := result["aggregations"].(map[string]interface{})["sum_field"].(map[string]interface{})["value"]
	return sumValue.(float64), nil
}

// 分组
func (p *Es) Group(field string) (map[string]interface{}, error) {
	aggregationQuery := map[string]interface{}{
		"size": 0,
		"aggs": map[string]interface{}{
			"group_field": map[string]interface{}{
				"terms": map[string]interface{}{
					"field": field,
					"size":  p.limit,
				},
			},
		},
	}
	for k, v := range p.query {
		aggregationQuery[k] = v
	}

	queryBytes, _ := json.Marshal(aggregationQuery)
	req := esapi.SearchRequest{
		Index: []string{p.indexName},
		Body:  bytes.NewBuffer(queryBytes),
	}
	res, err := req.Do(context.Background(), p.client)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()
	if res.IsError() {
		return nil, errors.New(res.String())
	}
	var result map[string]interface{}
	if err1 := json.NewDecoder(res.Body).Decode(&result); err1 != nil {
		return nil, err1
	}
	return result["aggregations"].(map[string]interface{})["group_field"].(map[string]interface{}), nil
}
